package com.atguigu.demo;

import org.apache.spark.SparkConf;
import org.apache.spark.sql.*;
import org.apache.spark.sql.expressions.Aggregator;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.function.BiConsumer;

import static org.apache.spark.sql.functions.udaf;
import static org.apache.spark.sql.functions.var_pop;

/**
 * @author yhm
 * @create 2022-12-26 15:03
 */
public class Test02_top3 {
    public static void main(String[] args) {

        // 修改系统用户名  获取执行权限
        System.setProperty("HADOOP_USER_NAME","atguigu");

        // 1. 创建sparkConf配置对象
        SparkConf conf = new SparkConf().setAppName("sql").setMaster("local[*]");

        // 2. 创建sparkSession连接对象
        // 开启hive的支持
        SparkSession spark = SparkSession.builder()
                .enableHiveSupport()
                .config(conf).getOrCreate();

        // 3. 编写代码
        // 一 过滤点击商品的数据
        Dataset<Row> dataset = spark.sql("select \n" +
                "  click_product_id,\n" +
                "  city_id\n" +
                "from user_visit_action uva\n" +
                "where click_product_id != -1");

        dataset.createOrReplaceTempView("t1");

        // 二 拼接字段  获取区域名称和商品名称
        spark.sql("select \n" +
                "    area,\n" +
                "    city_name,\n" +
                "    product_name\n" +
                "from t1\n" +
                "join city_info ci\n" +
                "on  t1.city_id = ci.city_id\n" +
                "join product_info pi\n" +
                "on  click_product_id=product_id").createOrReplaceTempView("t2");


        // 注册udaf函数
        spark.udf().register("cityMark",udaf(new CityMark(), Encoders.STRING()));

        // 三 统计区域商品点击次数
        // 统计区域内商品点击次数的同时   统计城市名称
        spark.sql("select \n" +
                "    area,\n" +
                "    product_name,\n" +
                "    cityMark(city_name) city_mark,\n" +
                "    count(*) click_nums\n" +
                "from t2\n" +
                "group by area,product_name").createOrReplaceTempView("t3");

        // 四 对区域内商品点击次数进行排序
        spark.sql("select \n" +
                "    area,\n" +
                "    product_name,\n" +
                "    click_nums,\n" +
                "    city_mark,\n" +
                "    rank()over(partition by area order by click_nums desc) rk\n" +
                "from t3").createOrReplaceTempView("t4");

        // 五 过滤出top3商品
        spark.sql("select \n" +
                "    area,\n" +
                "    product_name,\n" +
                "    click_nums,\n" +
                "    city_mark\n" +
                "from t4\n" +
                "where rk <= 3").show(false);

        // 4. 关闭sparkSession
        spark.close();
    }

    public static class Buffer implements Serializable {
        private Long  totalCount;
        private HashMap<String, Long> map;

        public Buffer() {
        }

        public Buffer(Long totalCount, HashMap<String, Long> map) {
            this.totalCount = totalCount;
            this.map = map;
        }

        public Long getTotalCount() {
            return totalCount;
        }

        public void setTotalCount(Long totalCount) {
            this.totalCount = totalCount;
        }

        public HashMap<String, Long> getMap() {
            return map;
        }

        public void setMap(HashMap<String, Long> map) {
            this.map = map;
        }
    }

    // IN 传入参数的类型
    // BUF 中间类型
    // OUT 输出类型
    public static class CityMark extends Aggregator<String,Buffer,String>{

        // 添加类 使用compare方法进行排序
        public static class CityCount {
            String cityName;
            Long cityCount;

            public CityCount() {
            }

            public CityCount(String cityName, Long cityCount) {
                this.cityName = cityName;
                this.cityCount = cityCount;
            }
        }

        public static class CompareCityCount implements Comparator<CityCount>{

            @Override
            public int compare(CityCount o1, CityCount o2) {
                if (o1.cityCount > o2.cityCount){
                    return -1;
                }else if (o1.cityCount.equals(o2.cityCount)){
                    return 0;
                }else {
                    return 1;
                }
            }
        }

        // 创建初始中间变量
        @Override
        public Buffer zero() {
            return new Buffer(0L,new HashMap<String, Long>());
        }

        @Override
        public Buffer reduce(Buffer b, String a) {
            // 累加获取的城市名称
            // 总次数+1
            b.setTotalCount(b.getTotalCount() + 1);

            // 对应的map里面  如果城市存在  次数+1  不存在 次数改为1
            HashMap<String, Long> map = b.getMap();
            map.put(a,map.getOrDefault(a,0L) + 1);
            return b;
        }

        @Override
        public Buffer merge(Buffer b1, Buffer b2) {
            b1.setTotalCount(b1.getTotalCount() + b2.getTotalCount());
            // 合并两个map
            HashMap<String, Long> map1 = b1.getMap();
            HashMap<String, Long> map2 = b2.getMap();

            map2.forEach(new BiConsumer<String, Long>() {
                @Override
                public void accept(String key, Long value) {
                    map1.put(key,map1.getOrDefault(key,0L) + value);
                }
            });

            return b1;
        }

        // buffer -> 总计100  map(北京->20,天津->30.. )
        @Override
        public String finish(Buffer reduction) {
            // 因为最终的字符串只保留前两个城市  后面的使用其他  所有需要对数据进行排序
            // 排序一个map
            HashMap<String, Long> map = reduction.getMap();
            ArrayList<CityCount> cityCounts = new ArrayList<>();
            map.forEach(new BiConsumer<String, Long>() {
                @Override
                public void accept(String s, Long aLong) {
                    cityCounts.add(new CityCount(s,aLong));
                }
            });
            cityCounts.sort(new CompareCityCount());

            // 拼接字符串的结果
            ArrayList<String> resultString = new ArrayList<>();
            Long totalCount = reduction.getTotalCount();
            Double sum=0.0;

            while (resultString.size() < 2 && cityCounts.size()>0){
                CityCount cityCount = cityCounts.get(0);
                double v = (cityCount.cityCount.doubleValue() / totalCount) * 100;
                sum += v;
                resultString.add(cityCount.cityName + String.format("%.2f", v) + "%");
                cityCounts.remove(0);
            }

            // 如果结果中还存在其他城市
            if (cityCounts.size()>0){
                resultString.add("其他" + String.format("%.2f", (100-sum)) + "%" );
            }

            StringBuilder stringBuilder = new StringBuilder();
            for (String s : resultString) {
                stringBuilder.append(s).append(",");
            }
            String s = stringBuilder.toString();
            return s.substring(0,s.length()-1);
        }

        @Override
        public Encoder<Buffer> bufferEncoder() {
            return Encoders.javaSerialization(Buffer.class);
        }

        @Override
        public Encoder<String> outputEncoder() {
            return Encoders.STRING();
        }
    }
}
